package mq

import (
	"fmt"
	"io"
	"tcptemplate/util"

	"github.com/pkg/errors"
	"gopkg.in/redis.v3"
	"qiniupkg.com/x/log.v7"

	MQTT "github.com/eclipse/paho.mqtt.golang"
)

type CallBack func(string, string)

//通用mq库，支持发布，订阅
type MqClient interface {
	Subscribe(topic string, callback *CallBack) error
	Public(topic string, data string) error
}

type RedisMqClient struct {
	client   *redis.Client
	Address  string
	Password string
}
type MqttClient struct {
	client   MQTT.Client
	Address  string
	ClientId string
	Options  *MQTT.ClientOptions
}

func (this *MqttClient) Connect(ip string, port int64, id string) error {
	this.Address = ip

	opts := MQTT.NewClientOptions().AddBroker(ip)
	opts.SetClientID(id)

	this.client = MQTT.NewClient(opts)
	if token := this.client.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}

	return nil
}

func (this *RedisMqClient) Connect(ip string, port int64, passwd string) error {
	option := &redis.Options{
		Addr:     ip,
		Password: passwd,
		DB:       1,
	}
	this.Address = ip
	this.Password = passwd
	this.client = redis.NewClient(option)
	// 检测client有效性
	if nil != this.client {
		_, err := this.client.Ping().Result()
		if nil != err {
			this.client.Close()
			return err
		}
	} else {
		return nil
	}
	return nil
}
func (this *MqttClient) Subscribe(topic string, callback CallBack) error {
	handler := func(c MQTT.Client, msg MQTT.Message) {
		msg.Payload()
		msg.Topic()
		callback(msg.Topic(), string(msg.Payload()))
	}
	opts := MQTT.NewClientOptions().AddBroker(this.Address)
	opts.SetClientID(this.ClientId)
	opts.SetDefaultPublishHandler(handler)

	this.Options = opts

	this.client = MQTT.NewClient(opts)
	if token := this.client.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}
	this.client.Subscribe(topic, 0, handler)
	return nil
}
func (this *RedisMqClient) Subscribe(topic string, callback CallBack) error {
	if nil == this {
		return errors.New(util.ERR_NULL_POINTER)
	}
	pubsub, err := this.client.Subscribe(topic)
	if nil != err {
		return err
	}
	x := make(chan bool, 1)
	go func() {
		x <- true
		defer pubsub.Close()
		for true {
			msg, err := pubsub.ReceiveMessage()
			if nil != err {
				log.Print(err)
				if err == io.EOF {
					return
				}
			}
			callback(topic, msg.Payload)
		}
	}()
	<-x
	return nil
}

// 创建redis对象
func creatRedisClient(option *redis.Options) (*redis.Client, error) {

	client := redis.NewClient(option)
	// 检测client有效性
	if nil != client {
		_, err := client.Ping().Result()
		if nil != err {
			client.Close()
			return nil, errors.New(fmt.Sprintf("fail to ping redis-svr,addr :%s , pwd :%s ,DB :%d", option.Addr, option.Password, option.DB))
		}
	} else {
		return nil, errors.New(fmt.Sprintf("fail to connect redis-svr,,addr :%s , pwd :%s ,DB :%d", option.Addr, option.Password, option.DB))
	}
	return client, nil
}

func (this *RedisMqClient) Public(topic string, data string) error {
	if nil == this {
		return errors.New(util.ERR_NULL_POINTER)
	}
	// 检测client有效性
	if nil != this.client {
		_, err := this.client.Ping().Result()
		if nil != err {
			this.client.Close()
			// 尝试3次重连
			for i := 0; i < 3; i++ {
				this.client, err = creatRedisClient(&redis.Options{
					Addr:     this.Address,
					Password: this.Password,
				})
				if this.client != nil {
				}
			}
		}
	}
	_, err := this.client.Publish(topic, data).Result()
	if nil != err {
		log.Print(err)
		return err
	}
	return nil
}

func (this *MqttClient) Public(topic string, data string) error {
	if nil == this {
		return errors.New(util.ERR_NULL_POINTER)
	}
	// 检测client有效性
	if true != this.client.IsConnected() {

	}
	token := this.client.Publish(topic, 0, false, data)
	token.Wait()
	if nil != token.Error() {
		log.Print(token.Error())
		return token.Error()
	}
	return nil
}
